/**
 * 
 */
package com.testing.mq.rabbitmq;

import java.io.IOException;
import java.nio.charset.Charset;

import com.rabbitmq.client.QueueingConsumer;

/**
 * RabbitMQ Consumer
 *
 * @author Jason
 * created at: Sep 26, 2011
 */
public class SimpleMQConsumer extends AbstractSimpleMQWorker {
	QueueingConsumer qConsumer;
	
	/**
	 * ctor.
	 * @param pQName
	 */
	public SimpleMQConsumer() {
		super();
	}
	/**
	 * ctor
	 * @param pQName
	 * @param pQHost
	 */
	public SimpleMQConsumer(String pQHost, String pQName) {
		super(pQHost, pQName);
	}

	/**
	 * init method; to be called right after ctor
	 */
	public void init() {
		super.init();
		
		this.qConsumer=new QueueingConsumer(this.qChannel);
		try {
			/*
			 * this way of consumption REQUIRES explicit 
			 * message Acknowledgment
			 */
			this.qChannel.basicConsume(this.qName, qConsumer);
			
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException("[SimpleMQConsumer][init]", e);
		}
	}

	/**
	 * Consume 1 message exactly, the timeout is 100ms
	 * @return
	 */
	public String consumeMessage() {
		StringBuilder aSB=new StringBuilder(100);
		
		try {
			QueueingConsumer.Delivery aDelivery=this.qConsumer.nextDelivery(1000);
			
			if (aDelivery!=null) {
				aSB.append("Message content~ ").
					append(new String(aDelivery.getBody(), Charset.forName("UTF-8"))).
					append("\r\n");
				/*
				 * if nothing goes wrong, send an acknowledge back to RabbitMQ, 
				 * so this message is said to be consumed
				 */
				this.qChannel.basicAck(aDelivery.getEnvelope().getDeliveryTag(), false);
			}
			
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("[SimpleMQConsumer][consumeMessage]", e);
		}
		return aSB.toString();
	}
	
	/**
	 * 
	 */
	@Override
	public void dispose() {
		if (this.qConsumer!=null) this.qConsumer=null;
		if (this.qChannel!=null) try { this.qChannel.close(); } catch(Exception e) {}
		if (this.qConnection!=null) try { this.qConnection.close(); } catch(Exception e) {}
	}
	
	
	
}
